package com.smile.realtime.app


import com.alibaba.fastjson.{JSON, JSONObject}
import com.smile.realtime.bean.DauInfo
import com.smile.realtime.utils.{MyESUtil, MyKafkaUtil, MyRedisUtil, OffsetManagerUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.lang
import java.util.Date
import java.text.SimpleDateFormat
import scala.collection.mutable.ListBuffer

/**
 * Author: smile
 * Date: 2022/2/11
 * Desc: 日活业务
 */
object DauApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("DauApp")
    val ssc: StreamingContext = new StreamingContext(conf,Seconds(3))
    var recordDStream: InputDStream[ConsumerRecord[String, String]] = null
    //通过SparkStreaming从Kafka中读取数据
    var topic: String = "smile_start_bak"
    var groupId: String = "mbw_mr"
    //通过偏移量工具从Redis中获取偏移量并进行判断是否为空
    val partitionToLong: Map[TopicPartition, Long] = OffsetManagerUtil.getOffset(topic, groupId)
    if(partitionToLong != null){
      //如果Redis中存在偏移量那么从Kafka偏移量位置进行消费
      recordDStream = MyKafkaUtil.getKafkaStream(topic, ssc, partitionToLong, groupId)
    }else {
      recordDStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }
    //得到本批次中处理数据的分区对应的偏移量起始及结束位置
    // 注意：这里我们从 Kafka 中读取数据之后，直接就获取了偏移量的位置，因为 KafkaRDD 可以转换为HasOffsetRanges，会自动记录位置
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]
    val offsetDStream: DStream[ConsumerRecord[String, String]] = recordDStream.transform {
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        println(offsetRanges(0).untilOffset + "*****")
        rdd
      }
    }
      val jsonObjDStream: DStream[JSONObject] = offsetDStream.map {
      record =>
      //获取启动日志
      val jsonStr: String = record.value()
      //将启动日志转换为 json 对象
      val jsonObj: JSONObject = JSON.parseObject(jsonStr)
      //获取时间戳 毫秒数
      val ts: Long = jsonObj.getLong("ts")
      //获取字符串 日期 小时
      val dateHourString: String = new SimpleDateFormat("yyyy-MM-dd HH").format(new Date(ts))
      //对字符串日期和小时进行分割，分割后放到 json 对象中，方便后续处理
      val dateHour: Array[String] = dateHourString.split(" ")
      jsonObj.put("dt",dateHour(0))
      jsonObj.put("hr",dateHour(1))
      jsonObj
    }
    // jsonObjDStream.print(1000)
    //通过Redis 对采集到的日志进行当天去重操作
    val filteredDStream: DStream[JSONObject] = jsonObjDStream.filter {
      jsonObject => {
        //获取登录日期
        var dt = jsonObject.getString("dt")
        val daukey = "daukey:" +  dt
        //获取设备id
        val myid = jsonObject.getJSONObject("common").getString("mid")
        //获取redis客户端
        val redisclient: Jedis = MyRedisUtil.getJedisClient
        val isFirst: lang.Long = redisclient.sadd(daukey, myid)
        //设置key的失效时间为24小时
        if (redisclient.ttl(daukey) < 0) {
          redisclient.expire(daukey, 3600 * 24)
        }
        //关闭redis
        redisclient.close()
        //从redis中判断是否已经登录过
        if (isFirst == 1L) {
          //第一次登录
          true
        } else {
          false
        }
      }
    }
    //通过redis采集到的日志进行去重，以分区为单位对数据进行处理，每一个分区获得一次redis连接
    val filteredDStream2: DStream[JSONObject] = jsonObjDStream.mapPartitions {
      jsonObjItr => {
        //获取 Redis 客户端
        val jedisClient: Jedis = MyRedisUtil.getJedisClient
        //定义当前分区过滤后的数据
        val filteredList: ListBuffer[JSONObject] = new ListBuffer[JSONObject]
        for (jsonObj <- jsonObjItr) {
          //获取当前日期
          val dt: String = jsonObj.getString("dt")
          //获取设备 mid
          val mid: String = jsonObj.getJSONObject("common").getString("mid")
          //拼接向 Redis 放的数据的 key
          val dauKey: String = "dau:" + dt
          //判断 Redis 中是否存在该数据
          val isNew: lang.Long = jedisClient.sadd(dauKey, mid)
          //设置key的失效时间为24小时
          if (jedisClient.ttl(dauKey) < 0) {
            jedisClient.expire(dauKey, 3600 * 24)
          }
          if (isNew == 1L) {
            //如果 Redis 中不存在，那么将数据添加到新建的 ListBuffer 集合中，实现过滤的效果
            filteredList.append(jsonObj)
          }
        }
        jedisClient.close()
        filteredList.toIterator
      }
    }
    //输出测试 数量会越来越少，最后变为 0 因为我们 mid 只是模拟了 50 个
    //filteredDStream2.count().print()

    //===============向 ES 中保存数据===================
    filteredDStream.foreachRDD{
      rdd=>{//获取 DS 中的 RDD
        rdd.foreachPartition{//以分区为单位对 RDD 中的数据进行处理，方便批量插入
          jsonItr=>{
            val dauList: List[(String,DauInfo)] = jsonItr.map {
              jsonObj => {
                //每次处理的是一个 json 对象 将 json 对象封装为样例类
                val commonJsonObj: JSONObject = jsonObj.getJSONObject("common")
                val dauInfo: DauInfo = DauInfo(
                  commonJsonObj.getString("mid"),
                  commonJsonObj.getString("uid"),
                  commonJsonObj.getString("ar"),
                  commonJsonObj.getString("ch"),
                  commonJsonObj.getString("vc"),
                  jsonObj.getString("dt"),
                  jsonObj.getString("hr"),
                  "00", //分钟我们前面没有转换，默认 00
                  jsonObj.getLong("ts")
                )
                (dauInfo.mid,dauInfo)
              }
            }.toList
            //对分区的数据进行批量处理
            //获取当前日志字符串
            val dt = new SimpleDateFormat("yyyy-MM-dd").format(new Date())
            MyESUtil.bulkInsert(dauList,"smile1483_dau_info_" + dt)
          }
        }
        OffsetManagerUtil.saveOffset(topic,groupId,offsetRanges)
      }
    }
    ssc.start()
    ssc.awaitTermination()
  }

}
